package com.navinfo.opentsp.platform.computing.analysis.application;

import com.mongodb.spark.MongoSpark;
import com.mongodb.spark.config.ReadConfig;
import com.mongodb.spark.rdd.api.java.JavaMongoRDD;
import com.navinfo.opentsp.platform.computing.analysis.pb.RealTimeDataPbOld;
import com.navinfo.opentsp.platform.computing.analysis.util.GpsTransform;
import com.navinfo.opentsp.platform.computing.analysis.util.PropertiesUtil;
import com.navinfo.opentsp.platform.location.protocol.common.RealTimeDataPb;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.SparkSession;
import org.bson.Document;
import org.bson.types.Binary;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import scala.Tuple2;

import java.text.SimpleDateFormat;
import java.util.*;

/**
 * @author chen jie
 * Created by chen on 2019/7/29.
 * 密集历史数据迁移
 */
public class MongoToKafkaD004 {

    public static void main(String[] args) {
        String day = args[0];
//        String day = "20190501";
        String mongoUri_D004 = PropertiesUtil.getProperties("mongoUri_D004");
        SparkSession spark = SparkSession.builder()
                .appName("MongoToKafkaD004")
//                .master("local")
                .config("spark.mongodb.input.uri",mongoUri_D004)
                .config("spark.mongodb.input.partitioner","MongoPaginateBySizePartitioner")
                .config("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
                .getOrCreate();
        try{
            JavaSparkContext jsc = new JavaSparkContext(spark.sparkContext());
            exec(day,jsc);
        }
        catch (Exception e){
            System.out.println(e.getMessage());
        }
        finally {
            spark.stop();
        }
    }

    private static void exec(String day, JavaSparkContext jsc) {
        Map<String, String> readOverrides = new HashMap<>();
        String topic = PropertiesUtil.getProperties("d004.topic");
        String kafka = PropertiesUtil.getProperties("kafka");
        readOverrides.put("collection","concentrated_data_"+day.substring(0,6));
        ReadConfig readConfig = ReadConfig.create(jsc).withOptions(readOverrides);
//        {beginDate:{$gte:1557279186000,$lte:1557379669000}}
//        Document filter = Document.parse("{ $match: {day : '"+ day + "'} }
        long start = 0L;
        long end = 0L;
        try {
            String res;
            SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyyMMdd");
            Date date = simpleDateFormat.parse(day);
            start = date.getTime();
            end = start+ 1000*60*60*24;

        }catch(Exception e){

        }
        Document filter = Document.parse("{ $match: {'beginDate': {$gte:"+start+",$lte:"+end+"}}}");
        JavaMongoRDD<Document> rdd = MongoSpark.load(jsc,readConfig).withPipeline(Collections.singletonList(filter));
        System.out.println(day +"数据量================"+rdd.count());
        rdd.mapPartitionsToPair((PairFlatMapFunction< Iterator<Document>,String, List<Binary>>) (Iterator<Document> itor) -> {
            List<Tuple2<String, List<Binary>>> ret = new ArrayList<>();
            while(itor.hasNext()){
                Document doc = itor.next();
                String tid = doc.getString("terminalId");
                Document data = (Document) doc.get("data");
                ArrayList<Binary> list = new ArrayList(data.values());
                ret.add(Tuple2.apply(tid, list));
            }
            return ret.iterator();
        }).reduceByKey((Function2<List<Binary>, List<Binary>, List<Binary>>) (elements1, elements2) -> {
            elements1.addAll(elements2);
            return elements1;
        }).foreachPartition(new VoidFunction<Iterator<Tuple2<String, List<Binary>>>>() {

            @Override
            public void call(Iterator<Tuple2<String, List<Binary>>> itor) throws Exception {
                Map<String, Object> props = new HashMap<>();
                props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka);
//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.30.50.21:9092");
                props.put(ProducerConfig.RETRIES_CONFIG, 0);
                props.put(ProducerConfig.BATCH_SIZE_CONFIG, 100000);
                props.put(ProducerConfig.LINGER_MS_CONFIG, 1);
                props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
                props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.navinfo.opentsp.platform.computing.analysis.entity.ProtoBufSerializer");
                props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
                KafkaTemplate<String,Object> kafkaTemplate = new KafkaTemplate<String, Object>(new DefaultKafkaProducerFactory<>(props));
                while(itor.hasNext()) {
                    Tuple2<String, List<Binary>> tuple2 = itor.next();
                    String tid = tuple2._1;
                    List<Binary> dataList = tuple2._2;
                    for (Binary binary : dataList) {
                        RealTimeDataPbOld.RealTimeData realTimeData = null;
                        try {
                            realTimeData = RealTimeDataPbOld.RealTimeData.parseFrom(binary.getData());
                        } catch (Exception e) {
                            e.printStackTrace();
                            continue;
                        }
                        //新老pb转换
                        RealTimeDataPb.RealTimeData.Builder builder = RealTimeDataPb.RealTimeData.newBuilder();
                        // 纠偏处理
                        int[] dataEncrypt = GpsTransform.transform(((double) realTimeData.getLongitude()) / 1000000, ((double) realTimeData.getLatitude()) / 1000000);
                        builder.setLongitude(dataEncrypt[0]);
                        builder.setLatitude(dataEncrypt[1]);
                        builder.setGpsTime(realTimeData.getGpsTime());
                        builder.setHeight(realTimeData.getHeight());
                        builder.setEngineOutputTorque(realTimeData.getEngineOutputTorque());
                        builder.setSpeed(realTimeData.getSpeed());
                        builder.setAccelerator(realTimeData.getAccelerator());
                        builder.setBrake(realTimeData.getBrake());
                        builder.setRotation(realTimeData.getRotation());
                        builder.setGear(realTimeData.getGear());
                        builder.setClutchSwitch(realTimeData.getClutchSwitch());
                        builder.setRealTimeOilConsumption(realTimeData.getRealTimeOilConsumption());
                        builder.setFuelConsumptionRate(realTimeData.getFuelConsumptionRate());
                        builder.setRetransmissionPackets(realTimeData.getRetransmissionPackets());
                        builder.setTerminalId(Long.parseLong(tid));
                        builder.setOriginalLat(realTimeData.getLatitude());
                        builder.setOriginalLng(realTimeData.getLongitude());
                        builder.setEngineBrakingState(realTimeData.getEngineBrakingState());
                        builder.setExhaustBrakingState(realTimeData.getExhaustBrakingState());
                        builder.setEmptySignal(realTimeData.getEmptySignal());
                        builder.setExhaustFlow(realTimeData.getExhaustFlow());
                        builder.setFanSpeed(realTimeData.getFanSpeed());
                        builder.setMechanicalBoxGear(realTimeData.getMechanicalBoxGear());
                        builder.setInstrumentSpeed(realTimeData.getInstrumentSpeed());
                        builder.setWheelSpeed(realTimeData.getWheelSpeed());
                        builder.setParkingBrakeSwitch(realTimeData.getParkingBrakeSwitch());
                        builder.setCruiseControlSettingSwitch(realTimeData.getCruiseControlSettingSwitch());
                        builder.setTargetGear(realTimeData.getTargetGear());
                        builder.setSlope(realTimeData.getSlope());
                        builder.setLoad(realTimeData.getLoad());
                        builder.setFuelLevel(realTimeData.getFuelLevel());
                        builder.setWaterTemperature(realTimeData.getWaterTemperature());
                        builder.setAtmosphericPressure(realTimeData.getAtmosphericPressure());
                        builder.setIntakeAirTemperature(realTimeData.getIntakeAirTemperature());
                        builder.setAtmosphericTemperature(realTimeData.getAtmosphericTemperature());
                        builder.setExhaustGasTemperature(realTimeData.getExhaustGasTemperature());
                        builder.setIntakeManifoldBoostPressure(realTimeData.getIntakeManifoldBoostPressure());
                        builder.setRelativePressure(realTimeData.getRelativePressure());
                        builder.setEngineTorqueMode(realTimeData.getEngineTorqueMode());
                        builder.setOilPressure(realTimeData.getOilPressure());
                        builder.setUreaLevel(realTimeData.getUreaLevel());
                        builder.setStatusFlag(realTimeData.getStatusFlag());
                        builder.setBrakeRate(realTimeData.getBrakeRate());
                        builder.setGpsDirection(realTimeData.getGpsDirection());
                        builder.setAirConditionStatus(realTimeData.getAirConditionStatus());
                        builder.setTransmissionRotation(realTimeData.getTransmissionRotation());
                        builder.setFrictionTorque(realTimeData.getFrictionTorque());
                        builder.setTerminalVersion(realTimeData.getVersion());
                        kafkaTemplate.send(topic, tid + "", builder.build().toByteArray());

                    }
                }
            }
        });
    }
}
